package com.test.producer;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

public class Producer {
	
	private String producerName;
	
	private String namesrvAddr;
	
	private Set<String> topicSet;
	
	private int interval;
	
	private String path;
	
	private Map<String, ScheduledExecutorService> wokers;
	
	public Producer(String filePath) throws IOException {
		super();
		this.topicSet = new HashSet<String>();
		this.wokers = new HashMap<String, ScheduledExecutorService>();
		this.path = filePath;
		Properties properties = load(path);
		this.producerName = properties.getProperty("producerName").toString();
		this.namesrvAddr = properties.getProperty("namesrvAddr").toString();
		this.interval = Integer.valueOf(properties.getProperty("interval").toString());
		String topicList = properties.getProperty("topicList").toString();
		if (topicList != null && topicList.length() > 0) {
			String[] split = topicList.split("\\,");
			for (String topic : split) {
				topicSet.add(topic);
			}
		}
	}

	DefaultMQProducer defaultMQProducer;
	
	public void start() throws MQClientException {
		defaultMQProducer = new DefaultMQProducer(producerName);
        
        defaultMQProducer.setNamesrvAddr(namesrvAddr);
        defaultMQProducer.setInstanceName(String.valueOf(System.currentTimeMillis()));
        //defaultMQProducer.setVipChannelEnabled(false);
        defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
        defaultMQProducer.start();
        initWorker();
        checkConfig();
        System.out.println(producerName + " start success!");
	}
	
	private void initWorker() {
		if (topicSet != null && topicSet.size() > 0) {
			for (String topic : topicSet) {
				addWorker(topic);
			}
		}
	}
	
	public void addWorker(final String topic) {
		if (!wokers.containsKey(topic)) {
			ScheduledExecutorService worker = Executors.newScheduledThreadPool(10);
			wokers.put(topic, worker);
			worker.scheduleAtFixedRate(new Runnable() {
				
				public void run() {
					String key = UUID.randomUUID().toString();
					String content = "Topic: " + topic + " ,key: " + key + " ,timestamp; " + System.currentTimeMillis();
					try {
						System.out.println("Send message -->" + content);
						defaultMQProducer.send(new Message(topic, topic, key, content.getBytes()));
					} catch (MQClientException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (RemotingException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (MQBrokerException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}, 1000*10, interval, TimeUnit.MICROSECONDS);
		}
	}
	
	public void checkConfig() {
		ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
		scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
			
			public void run() {
				try {
					Properties properties = load(path);
					String topicList = properties.get("topicList").toString();
					Set<String> topics = new HashSet<String>();
					if (topicList != null && topicList.length() > 0) {
						String[] split = topicList.split("\\,");
						for (String topic : split) {
							topics.add(topic);
						}
					}
					
					for (Map.Entry<String, ScheduledExecutorService> entry : wokers.entrySet()) {
						String topic = entry.getKey();
						ScheduledExecutorService woker = entry.getValue();
						if (!topics.contains(topic)) {
							woker.shutdown();
							wokers.remove(topic);
						}
					}
					
					for (String topic : topics) {
						if (!wokers.containsKey(topic)) {
							addWorker(topic);
						}
					}
					
					
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}, 10, 30, TimeUnit.SECONDS);
	}

	private Properties load(String filePath) throws IOException {
		Properties properties = new Properties();
		InputStream in = null;
		try {
			if (filePath == null) {
				in = ProducerMain.class.getResourceAsStream("/config.properties");
			} else {
				in = new FileInputStream(filePath);
			}
			properties.load(in);
			return properties;
		} finally {
			if (in != null) {
				in.close();
			}
		}
	}
}
